package hust.rpc.client;


import hust.rpc.procotol.JsonRpcRequest;
import hust.rpc.util.JsonUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.pool.AbstractChannelPoolMap;
import io.netty.channel.pool.ChannelPoolHandler;
import io.netty.channel.pool.FixedChannelPool;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.json.JsonObjectDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.net.InetSocketAddress;
import java.util.concurrent.CompletableFuture;

public class NettyClient {

    private static final Logger LOGGER = LogManager.getLogger(NettyClient.class);

    private AbstractChannelPoolMap<InetSocketAddress , FixedChannelPool> poolMap;

    private Bootstrap bootstrap;

    private NioEventLoopGroup group;

    private volatile static NettyClient client;

    private UnprocessedRequests unprocessedRequests = UnprocessedRequests.getUnprocessedRequests();

    private NettyClient() {
        init();
    }

    private void init() {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();

        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE,true);

        poolMap = new AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool>() {
            @Override
            protected FixedChannelPool newPool(InetSocketAddress inetSocketAddress) {
                LOGGER.info("Create new FixedChannelPool");
                return new FixedChannelPool(bootstrap.remoteAddress(inetSocketAddress), new ChannelPoolHandler() {
                    @Override
                    public void channelReleased(Channel ch) throws Exception {
                        LOGGER.info("channel released");
                    }

                    @Override
                    public void channelAcquired(Channel ch) throws Exception {
                        LOGGER.info("channel acquired");
                    }

                    @Override
                    public void channelCreated(Channel ch) throws Exception {
                        LOGGER.info("channel Created");

                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new JsonObjectDecoder());
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(new StringEncoder());

                        pipeline.addLast(new NettyClientHandler());
                    }
                },20);//单个host连接池大小
            }
        };
    }

    public static NettyClient getInstance(){
        if (client == null){
            synchronized (NettyClient.class){
                if (client == null){
                    client = new NettyClient();
                }
            }
        }
        return client;
    }

    public Object send(InetSocketAddress inetSocketAddrss, JsonRpcRequest request) {

        FixedChannelPool pool = poolMap.get(inetSocketAddrss);

        Future<Channel> future = pool.acquire();

        CompletableFuture<String> resultFuture = new CompletableFuture<>();

        future.addListener(new FutureListener<Channel>(){

            @Override
            public void operationComplete(Future<Channel> f) throws Exception {
                if (f.isSuccess()){
                    Channel channel = f.getNow();

                    channel.writeAndFlush(JsonUtil.writeRequestAsString(request));
                    unprocessedRequests.put(request.getId(),resultFuture);
                    pool.release(channel);
                }
            }
        });
        return resultFuture;
    }
}
